package com.itcj.dmp.report
import com.itcj.dmp.ConstantSchema
import com.itcj.dmp.report.RegionReportProcess.keys
import com.itcj.dmp.utils.KuduHelper
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types.{StringType, StructField, StructType}

object AdsRegionReportProcessor extends CommonReport {
  override def odsTablename: String = "ODS_" + KuduHelper.today()

  override def resultTableName: String = "ODS_AdsRegion"+ KuduHelper.today()

  override def keys: List[String] = List("region", "city")

  override def processDf(sourceDf: DataFrame): DataFrame = {
    import org.apache.spark.sql.functions._
    sourceDf.createOrReplaceTempView("sourceTableName")
    val kpiDF: DataFrame = sourceDf.sparkSession.sql(ConstantSchema.kpiSQL("sourceTableName"))
    kpiDF.createOrReplaceTempView("kpiDF")
    kpiDF.sparkSession.sql(ConstantSchema.rateSQL("kpiDF"))
  }

  override def schemaChange(schema: StructType): Unit = {
    for (key <- keys) {
      val keyindex = schema.fieldIndex(key)
      schema.fields.update(keyindex,new StructField(key, StringType, false))
    }
  }

  def main(args: Array[String]): Unit = {
   process()
  }
}
